package com.yh.csx.bsf.ods.monitor;

import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

import javax.annotation.PreDestroy;

import java.util.Optional;
import java.util.Set;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.SchedulingConfigurer;
import org.springframework.scheduling.config.CronTask;
import org.springframework.scheduling.config.ScheduledTaskRegistrar;
import org.springframework.scheduling.support.PeriodicTrigger;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import com.yh.csx.bsf.core.db.DbConn;
import com.yh.csx.bsf.core.util.ConvertUtils;
import com.yh.csx.bsf.core.util.LogUtils;
import com.yh.csx.bsf.core.util.PropertyUtils;
import com.yh.csx.bsf.ods.StragtegyProvider;
import com.yh.csx.bsf.ods.meta.OdsSyncAdapter;
import com.yh.csx.bsf.ods.utils.CRCUtils;
import com.yh.csx.bsf.ods.utils.WarnUtils;

import lombok.Data;
import lombok.var;

/**
 * 数据校验程序
 * 
 * @author Robin.Wang
 */
@Component
public class CheckSchedule implements SchedulingConfigurer {
	public static final String DATETIMEFORMAT="yyyy-MM-dd HH:mm:ss";
	/**
	 * 所有配置表
	 */
	public static volatile Map<String, String> SYNC_TABLE_MAPPING = new ConcurrentHashMap<String, String>();
	/**
	 * 所有任务
	 */
	public static Map<String, CheckTask> CHECK_TABLE_TASKS = new ConcurrentHashMap<String, CheckTask>();
	/**
	 * 任务注册管理器
	 */
	private volatile ScheduledTaskRegistrar registrar;
	private ScheduledFuture<?> futureTask;
	private CronTask task;
	/**
	 * 策略类
	 */
	@Autowired
	private StragtegyProvider stragtegyProvider;

	// 同步器
	@Autowired(required = false)
	Map<String, OdsSyncAdapter> syncers;

	/**
	 * 检查表
	 * 
	 * @throws Exception
	 */
	public void checktables() throws Exception {
		LogUtils.warn(this.getClass(), "ods", "【数据校验任务启动】。。。");
		for (Entry<String, CheckTask> checkTask : CHECK_TABLE_TASKS.entrySet()) {	
			try {
				checkCount(checkTask.getValue());
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
		StringBuilder sb = new StringBuilder();
		for (Entry<String, CheckTask> checkTask : CHECK_TABLE_TASKS.entrySet()) {
			if (Math.abs(checkTask.getValue().getSumDiffCount()) > PropertyUtils
					.getPropertyCache("bsf.health.strategy.sync.tables.diffcount", 100)) {
				sb.append("[").append(checkTask.getValue().getTargetTable()).append("]").append("条数差异")
						.append(checkTask.getValue().getDiffCount());
			}
			if (Math.abs(Optional.ofNullable(checkTask.getValue().getContentDiffCount()).orElse(0L)) > PropertyUtils
					.getPropertyCache("bsf.health.strategy.sync.tables.diffcount", 50)) {
				sb.append("内容差异条数").append(checkTask.getValue().getDiffCount());
			}
			sb.append("\n");
		}
		if (sb.length() > 0) {
			WarnUtils.alarm("数据校验不一致", sb.toString(), null);
		}
	}

	/**
	 * 检查数量
	 * 
	 * @throws Exception
	 */
	public void checkCount(CheckTask checktask) throws Exception {		
		Calendar c=Calendar.getInstance();
		c.add(Calendar.MINUTE,-PropertyUtils.getPropertyCache("bsf.health.strategy.sync.tables.delay", 5));
		String checkTime=new SimpleDateFormat(DATETIMEFORMAT).format(c.getTime());
		if (syncers != null) {
			String sqlFull="select count(*) as count from #{datatable}";
			String sql5minCount = "select count(*) as count from #{datatable} where update_time> '"+checkTime+"'";
			long sourceSumCount = 0;
			long last5minCount=0;
			for (CheckSubTask subTask : checktask.subTasks.values()) {
				try (DbConn dbConn = stragtegyProvider.getDb(subTask.getSourceDb())) {
					if (dbConn != null) {
						ResultSet rt = dbConn.executeResultSet(sqlFull.replace("#{datatable}", subTask.getSourceDb() + "." + subTask.getSourceTable()),	null);
						if (rt.next()) {
							sourceSumCount += Optional.ofNullable(rt.getInt("count")).orElse(0);
						}
					}
				} catch (Exception e) {					
					LogUtils.warn(this.getClass(), "ods", "【数据校验】SQL执行异常:" + sqlFull,e);
					return;
				}
				try (DbConn dbConn = stragtegyProvider.getDb(subTask.getSourceDb())) {
					if (dbConn != null) {
						ResultSet rt = dbConn.executeResultSet(sql5minCount.replace("#{datatable}", subTask.getSourceDb() + "." + subTask.getSourceTable()),null);
						if (rt.next()) {
							last5minCount += Optional.ofNullable(rt.getInt("count")).orElse(0);
						}
					}
				} catch (Exception e) {
					LogUtils.warn(this.getClass(), "ods", "【数据校验】SQL执行异常:" + sql5minCount,e);
					return;
				}
			}
			if (syncers != null) {
				for (OdsSyncAdapter adapter : syncers.values()) {
					if (adapter.existsTable(checktask.targetTable)) {
						long sumCount = adapter.selectCount(sqlFull.replace("#{datatable}", checktask.getTargetTable()));
						long min5Count = adapter.selectCount(sql5minCount.replace("#{datatable}", checktask.getTargetTable()));
						LogUtils.info(this.getClass(), "ods","【数据据校验条数】 " + checktask.targetTable + "总差异:" + (sumCount-sourceSumCount));
						LogUtils.info(this.getClass(), "ods","【数据据校验条数】" + checktask.targetTable + "最近5分钟差异:" + (min5Count-last5minCount));
						checktask.setSumDiffCount(sumCount - sourceSumCount);
						checktask.setLast5minDiffCount(min5Count-last5minCount);
						checktask.setRelativeDiffCount(sumCount - sourceSumCount-(min5Count-last5minCount));
						checktask.setChecktime(new Date());						
						if(Math.abs(checktask.getRelativeDiffCount()) > PropertyUtils.getPropertyCache("bsf.health.strategy.sync.tables.diffcount", 100)) {
							WarnUtils.alarm(checktask.targetTable,	 "[数据据校验条数不一致,差异：" + checktask.getRelativeDiffCount()+"]", null);
						}						
					}
				}
			}
		}
	}

	/**
	 * 内容校验
	 */
	public void checkContent(CheckTask checktask) throws Exception {
		long diffCount = 0;
		if (syncers != null) {
			String srcSql = "select * from {dbtable} order by id desc  limit 50" + "	union "
					+ "	select * from {dbtable} order by id asc  limit 50" + "	union all"
					+ "	select * from {dbtable} where id> (select * from ((select ((max(id)-min(id))/2) from accounting_batch_create_log) b)) order by id desc limit 50";

			String targSql = "select *  from {dbtable} where id=?";
			for (CheckSubTask subTask : checktask.subTasks.values()) {
				try (DbConn dbConn = stragtegyProvider.getDb(subTask.getSourceDb())) {
					if (dbConn != null) {
						ResultSet rs = dbConn.executeResultSet(srcSql.replaceAll("\\{dbtable\\}",
								subTask.getSourceDb() + "." + subTask.getSourceTable()), null);
						ResultSetMetaData meta = rs.getMetaData();
						int colCount = meta.getColumnCount();
						long crcSum = 0;
						long crcSum2 = 0;
						while (rs.next()) {
							for (int i = 1; i <= colCount; i++) {
								String value = rs.getString(i);
								long crc = CRCUtils.mkCrc16(value);
								crcSum += crc;
							}
						}
						while (rs.next()) {
							for (OdsSyncAdapter adapter : syncers.values()) {
								ResultSet rt2 = adapter.query(targSql.replace("{dbtable}", checktask.targetTable)
										.replace("?", rs.getString("id")));

								while (rt2.next()) {
									for (int i = 1; i <= colCount; i++) {

										String value = rt2.getString(i);
										long crc = CRCUtils.mkCrc16(value);
										crcSum2 += crc;
									}
								}
							}
						}
						if (crcSum != crcSum2) {
							diffCount++;
						}
					}
				} catch (Exception e) {
					LogUtils.debug(this.getClass(), "ods", "【数据校验】SQL执行异常:" + srcSql);
				}
			}
			checktask.setContentDiffCount(diffCount);
			if (diffCount > PropertyUtils.getPropertyCache("bsf.health.strategy.sync.tables.diffcount", 50)) {
				WarnUtils.alarm(checktask.targetTable, "[数据内容不一致条数:" + diffCount + "条]", null);				
			}
		}
	}

	@Override
	public void configureTasks(ScheduledTaskRegistrar taskRegistrar) {
		this.registrar = taskRegistrar;
		// 每5s扫描一次任务列表
		this.registrar.addTriggerTask(() -> {
			this.refreshTasks();
		}, triggerContext -> new PeriodicTrigger(900L, TimeUnit.SECONDS).nextExecutionTime(triggerContext));
	}

	/**
	 * 刷新任务
	 */
	private void refreshTasks() {
		freshConfigs();
		String expression = PropertyUtils.getPropertyCache("bsf.check.cron", "0 */10 * * * ?");
		if (this.task != null && this.task.getExpression().equals(expression)) {
			return;
		}
		if (futureTask != null)
			futureTask.cancel(true);
		this.task = new CronTask(() -> {
			try {
				this.checktables();
			} catch (Exception e) {
				LogUtils.error(this.getClass(), "ods", "【数据校验任务】运行异常:", e);
			}

		}, expression);
		futureTask = registrar.getScheduler().schedule(this.task.getRunnable(), this.task.getTrigger());
	}

	@PreDestroy
	public void destroy() {
		this.registrar.destroy();
	}

	public synchronized void freshConfigs() {
		Map<String, String> newTasks = new ConcurrentHashMap<String, String>();
		for (var db : stragtegyProvider.getDbs()) {
			try (DbConn dbConn = stragtegyProvider.getDb(db)) {
				if (dbConn != null) {
					List<Map<String, Object>> rs = dbConn.executeList(
							"select distinct table_name from information_schema.columns where table_schema =? and (COLUMN_NAME=? or COLUMN_NAME=?)",
							new Object[] { db, "id", "ID" });
					for (var r : rs) {
						var table = ConvertUtils.convert(r.get("TABLE_NAME"), String.class);
						String targetTable=stragtegyProvider.getTargetTable(db, table);
						String blackTable=stragtegyProvider.getBlackTable(db, table);
						if(!StringUtils.hasText(blackTable)&&StringUtils.hasText(targetTable)) {
							newTasks.put(db + "." + table, targetTable);
						}
					}
				} else {
					LogUtils.info(this.getClass(), "ods", "数据源未配置:" + db);
				}

			} catch (Exception e) {
				LogUtils.error(this.getClass(), "ods", "数据任务初始化异常", e);
			}
		}
		
		for (Entry<String, String> entry : newTasks.entrySet()) {			
			SYNC_TABLE_MAPPING.put(entry.getKey(), entry.getValue());
		}
		for (Entry<String, String> entry : SYNC_TABLE_MAPPING.entrySet()) {
			if (!newTasks.containsKey(entry.getKey())) {
				SYNC_TABLE_MAPPING.remove(entry.getKey());
			}
		}
		Set<String> checkTables = CheckSchedule.SYNC_TABLE_MAPPING.keySet();
		for (String checkTable : checkTables) {
			String[] checkdbtable = checkTable.split("\\.");
			String targetTable = stragtegyProvider.getTargetTable(checkdbtable[0], checkdbtable[1]);
			if (StringUtils.hasText(targetTable)) {
				if (CHECK_TABLE_TASKS.containsKey(targetTable)) {
					CHECK_TABLE_TASKS.get(targetTable).getSubTasks().put(checkTable,new CheckSubTask(checkdbtable[0], checkdbtable[1], targetTable));
				} else {			
					CHECK_TABLE_TASKS.put(targetTable, new CheckTask(targetTable));
					CHECK_TABLE_TASKS.get(targetTable).getSubTasks().put(checkTable,new CheckSubTask(checkdbtable[0], checkdbtable[1],targetTable));							
				}
			}
		}
	}

	/**
	 * 校验任务
	 */
	@Data
	public static class CheckTask {

		public CheckTask(String targetTable) {
			super();
			this.targetTable = targetTable;
		}
		private Long sumDiffCount;
		private Long last5minDiffCount;
		private Long relativeDiffCount;
		private Long contentDiffCount;
		private String targetTable;
		private Date checktime;
		private Map<String, CheckSubTask> subTasks = new HashMap<String, CheckSubTask>();

		/**
		 * 合并计算
		 */
		public Integer getDiffCount() {
			int differCount = 0;
			for (CheckSubTask subTask : subTasks.values()) {
				differCount += Optional.ofNullable(subTask.getDiffCount()).orElse(0);
			}
			return differCount;
		}

	}

	/**
	 * 校验子任务类
	 */
	@Data
	public static class CheckSubTask {
		/**
		 * 原库
		 */
		private String sourceDb;
		/**
		 * 原表
		 */
		private String sourceTable;
		/**
		 * 目标表
		 */
		private String targetTable;

		/**
		 * 不一致
		 */
		private Integer diffCount;

		public CheckSubTask() {
			super();
		}

		public CheckSubTask(String sourceDb, String sourceTable,String targetTable) {
			super();
			this.sourceDb = sourceDb;
			this.sourceTable = sourceTable;
			this.targetTable = targetTable;
		}

	}	
}
